package net.wlm.realtime.new_job;

import com.alibaba.fastjson.JSON;
import net.wlm.realtime.bean.TripMetrics;
import net.wlm.realtime.bean.VehicleData;
import net.wlm.realtime.funcation.ItineraryAnalysisFunction;
import net.wlm.realtime.utils.JdbcUtil;
import net.wlm.realtime.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


import java.time.Duration;


/**
 * 车辆行程分析作业类，用于处理车辆实时数据并进行行程统计分析
 */
public class ItineraryAnalysisJob {
    /**
     * 主方法，执行车辆行程分析作业
     *
     * @param args 命令行参数
     * @throws Exception 执行过程中可能抛出的异常
     */
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从Kafka消费数据
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle-data");

        // 转换为VehicleData对象
        DataStream<VehicleData> vehicleDataStream = kafkaStream.map(json ->
                        JSON.parseObject(json, VehicleData.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VehicleData>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
                );

        // 行程分析处理
        // 可选: 添加会话窗口计算
        DataStream<TripMetrics> windowedMetrics = vehicleDataStream
                .keyBy(VehicleData::getVin)
                .window(EventTimeSessionWindows.withGap(Time.minutes(10)))  // 使用10分钟的会话间隔
                .process(new ItineraryAnalysisFunction());
        // windowedMetrics.print();
        SingleOutputStreamOperator<String> map = windowedMetrics.map(new MapFunction<TripMetrics, String>() {
            @Override
            public String map(TripMetrics value) throws Exception {
                // 将布尔值转换为1/0
                int speedingValue = value.getSpeeding() ? 1 : 0;
                String output = value.getVin() + ',' +
                                value.getStartTime() + ',' +
                                value.getEndTime() + ',' +
                                value.getDistance() + ',' +
                                value.getDuration() + ',' +
                                value.getAvgSpeed() + "," +
                                value.getMaxSpeed() + "," +
                                speedingValue;
                return output;
            }
        });
        //  插入数据
        String upsertSql = "INSERT INTO vehicle_real.trip_metrics " +
                "(vin, start_time, end_time, distance, duration, avg_speed, max_speed, speeding) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
        //  插入数据
        JdbcUtil.sinkToClickhouseUpsert(map, upsertSql);
        // 触发执行
        env.execute("");
    }

}